package queue

import (
	"context"
	"fmt"
	"log"
	"shop/internal/config"
	"sync"
	"time"

	"github.com/gin-gonic/gin"
	"github.com/zhufuyi/sponge/pkg/logger"
	"go.uber.org/zap"

	"github.com/hibiken/asynq"
)

const Prefix string = "tk:queue:"
const QName string = "admin"

// Order 队列名称集合
const (
	Order = Prefix + "order:deliver"
)

// CustomerProcessor queue下面所有的方法都需要实现的类
type CustomerProcessor interface {
	SendTask(params Payload) (*asynq.Task, error)
	ProcessTask(ctx context.Context, t *asynq.Task) error
}

// Payload 统一queue参数
type Payload struct {
	Params    interface{}   `json:"params,omitempty"`
	ProcessIn time.Duration `json:"process_in,omitempty"`
}

var (
	queue *Queue
	once1 sync.Once
)

// Queue 队列实例
type Queue struct {
	serverCtx context.Context
	client    *asynq.Client
	opt       asynq.RedisConnOpt
	logger    *zap.Logger
	taskId    *string
}

// InitQueue NewQueue 实例化queue
func InitQueue() *Queue {
	cfg := config.Get()
	var opt asynq.RedisConnOpt
	if cfg.App.Env == gin.ReleaseMode {
		opt = asynq.RedisClusterClientOpt{
			Addrs: []string{"127.0.0.1:6379"},
		}
	} else {
		opt = asynq.RedisClientOpt{
			Addr: "127.0.0.1:6379",
		}
	}

	queue = &Queue{
		serverCtx: context.Background(),
		client:    asynq.NewClient(opt),
		opt:       opt,
		logger:    logger.Get(),
	}
	return queue
}

// GetQueue  get db
func GetQueue() *Queue {
	if queue == nil {
		once1.Do(func() {
			InitQueue()
		})
	}

	return queue
}

// SetTaskID 设置任务ID
func (s Queue) SetTaskID(taskId *string) *Queue {
	s.taskId = taskId
	return &s
}

// StartServer 启动监听服务
func (s Queue) StartServer() {
	logger.Info("queue启动成功")
	srv := asynq.NewServer(
		s.opt,
		asynq.Config{
			// 每个进程并发执行的worker数量
			Concurrency: 10,
			// Optionally specify multiple queues with different priority.
			Queues: map[string]int{
				QName + "critical": 6,
				QName + "default":  3,
				QName + "low":      1,
			},
			// See the godoc for other configuration options
		},
	)
	mux := asynq.NewServeMux()
	// mux.HandleFunc(TypeEmailDelivery, HandleEmailDeliveryTask)
	background := context.Background()
	mux.Handle(Order, NewOrderProcessor(background))
	if err := srv.Run(mux); err != nil {
		log.Fatalf("could not run server: %v", err)
	}
}

// SendQueue 发送队列
func (s Queue) SendQueue(obj CustomerProcessor, params Payload) error {
	task, err := obj.SendTask(params)
	if err != nil {
		s.logger.Error("could not create task:", logger.Err(err))
	}

	var info *asynq.TaskInfo
	if s.taskId != nil && *s.taskId != "" {
		info, err = s.client.Enqueue(task, asynq.Queue(QName+"default"), asynq.TaskID(*s.taskId))
	} else {
		info, err = s.client.Enqueue(task, asynq.Queue(QName+"default"))
	}

	if err != nil {
		return err
	}
	fmt.Printf("taskInfo : %# v\n", info)

	return nil
}
